{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "182839e9",
   "metadata": {},
   "source": [
    "# Deploy an MCP Gateway with existing Ray Serve apps\n",
    "\n",
    "This guide shows how to put a front-end on your existing Ray Serve services with a single MCP gateway—no code changes required in your classifier or translator. This guide follows this multi-app deployment pattern: https://docs.ray.io/en/latest/serve/multi-app.html.\n",
    "\n",
    "## Why MCP Gateway on Anyscale?\n",
    "If you already have Ray Serve services running on Anyscale (or your own cluster), you can front‐end them with a single MCP Gateway without touching the code. Just write one new `mcp_gateway.py`, regenerate your `config.yaml`, and an Anyscale service automatically spins up:\n",
    "\n",
    "* **Independent scaling for each Serve app**: Each app keeps its own autoscaling, load balancing, CPU/GPU settings, etc.\n",
    "\n",
    "* **A unified, streamable HTTP endpoint (/mcp) that multiplexes calls**\n",
    "\n",
    "* **No need to merge codebases**: Your classifier still lives in `image_classifier.py`, translator in `text_translator.py`; MCP gateway sits in front.\n",
    "\n",
    "The following is the architecture diagram illustrating the MCP Gateway with Ray Serve and Anyscale service:\n",
    "\n",
    "<img\n",
    "  src=\"https://agent-and-mcp.s3.us-east-2.amazonaws.com/mcp/diagrams/mcp_gateway_ray_serve.png\"\n",
    "  alt=\"MCP Gateway with Ray Serve Architecture\"\n",
    "  style=\"width:45%; display: block; margin: 0 auto;\"\n",
    "/>\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dbf08c9e",
   "metadata": {},
   "source": [
    "## Prerequisites\n",
    "- Ray [serve], included in the base Docker image\n",
    "- MCP Python library \n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9e5194d4",
   "metadata": {},
   "source": [
    "### Dependencies\n",
    "**Install the required Python packages and Podman:**\n",
    "```bash\n",
    "pip install  mcp==1.11.0 asyncio==3.4.3 pydantic==2.9.2\n",
    "```\n",
    "\n",
    "**Alternative: Docker image for Ray Serve deployment**\n",
    "\n",
    "You can also [build a Docker image for deployment on Anyscale](https://docs.anyscale.com/configuration/dependency-management/dependency-byod/) using the [Dockerfile included in this code repo](./Dockerfile). \n",
    "\n",
    "**Note**\n",
    " This Docker image is provided solely to deploy the MCP with Ray Serve. \n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0de07bbf",
   "metadata": {},
   "source": [
    "## 1. Integrate with existing Ray Serve apps\n",
    "\n",
    "Assume you already have two Ray Serve services: an image classifier and a text translator. Both code files live in the `mcp-gateway-with-existing-ray-apps` folder:\n",
    "* [image_classifier.py](./mcp-gateway-with-existing-ray-apps/image_classifier.py)\n",
    "* [text_translator.py](./mcp-gateway-with-existing-ray-apps/text_translator.py)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3feccbca",
   "metadata": {},
   "source": [
    "Now, create a new file named `mcp_gateway.py` that uses FastMCP to bridge between these services. \n",
    "\n",
    "In that file, you can retrieve the image‐classifier handle:\n",
    "\n",
    "```\n",
    "clf = serve.get_deployment_handle(\"image_classifier\", app_name=\"image_classifier_app\")\n",
    "```\n",
    "\n",
    "Ensure that in your Serve config you name the deployment `image_classifier` and the application `image_classifier_app`. \n",
    "\n",
    "Similarly, you can retrieve the text-translator handle as follows:\n",
    "\n",
    "```\n",
    "tr = serve.get_deployment_handle(\"text_translator\", app_name=\"text_translator_app\")\n",
    "```\n",
    "\n",
    "Ensure that in your Serve config you name the deployment `text_translator` and the application `text_translator_app`. \n",
    "\n",
    "The config details are in the next step."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ceab9279",
   "metadata": {},
   "outputs": [],
   "source": [
    "from contextlib import asynccontextmanager\n",
    "import fastapi\n",
    "from ray import serve\n",
    "from mcp.server.fastmcp import FastMCP\n",
    "\n",
    "# --------------------------------------------------------------------------\n",
    "# 1.  Create FastMCP in stateless http (streamable) mode\n",
    "# --------------------------------------------------------------------------\n",
    "mcp = FastMCP(\"Image-N-Translate\", stateless_http=True)\n",
    "\n",
    "# --------------------------------------------------------------------------\n",
    "# 2.  Register your tools BEFORE mounting the app\n",
    "# --------------------------------------------------------------------------\n",
    "\n",
    "@mcp.tool()\n",
    "async def classify(image_url: str) -> str:\n",
    "    \"\"\"Return the top-1 label for an image URL.\"\"\"\n",
    "    clf = serve.get_deployment_handle(\"image_classifier\", app_name=\"image_classifier_app\")\n",
    "    return await clf.classify.remote(image_url)\n",
    "\n",
    "@mcp.tool()\n",
    "async def translate(text: str) -> str:\n",
    "    \"\"\"Translate English → French.\"\"\"\n",
    "    tr  = serve.get_deployment_handle(\"text_translator\", app_name=\"text_translator_app\")\n",
    "    return await tr.translate.remote(text)\n",
    "\n",
    "\n",
    "# ----------------------------------------------------------------------------\n",
    "# 3.  Build FastAPI app with lifespan to mount the FastMCP streamable HTTP app\n",
    "# ----------------------------------------------------------------------------\n",
    "@asynccontextmanager\n",
    "async def lifespan(app: fastapi.FastAPI):\n",
    "    # After startup, mount the streamable-http MCP app.\n",
    "    app.mount(\"/\", mcp.streamable_http_app())\n",
    "\n",
    "    # Keep MCP’s session manager running for the lifetime of this process.\n",
    "    async with mcp.session_manager.run():\n",
    "        yield\n",
    "\n",
    "api = fastapi.FastAPI(lifespan=lifespan)\n",
    "\n",
    "# --------------------------------------------------------------------------\n",
    "# 4.  Wrap in a Ray Serve deployment\n",
    "# --------------------------------------------------------------------------\n",
    "@serve.deployment(\n",
    "    autoscaling_config={\n",
    "        \"min_replicas\": 2,\n",
    "        \"max_replicas\": 10,\n",
    "        \"target_ongoing_requests\": 50,\n",
    "    },\n",
    "    ray_actor_options={\n",
    "        \"num_cpus\": 0.5\n",
    "    }\n",
    ")\n",
    "@serve.ingress(api)\n",
    "class MCPGateway:\n",
    "\n",
    "    def __init__(self):\n",
    "        pass  \n",
    "\n",
    "\n",
    "# --------------------------------------------------------------------------\n",
    "# 5.  Expose the Serve app graph\n",
    "# --------------------------------------------------------------------------\n",
    "app = MCPGateway.bind()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6ff55cf3",
   "metadata": {},
   "source": [
    "## 2. Compile the config file\n",
    "\n",
    "Go to the `mcp-gateway-with-existing-ray-apps` directory containing those 3 Python files and run the following command:\n",
    "\n",
    "```bash\n",
    "cd mcp-gateway-with-existing-ray-apps\n",
    "serve build image_classifier:app text_translator:app mcp_gateway:app -o config_serve_temp.yaml\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ac739b53",
   "metadata": {},
   "source": [
    "In the `config_serve_temp.yaml` file, application names are auto-generated by default (e.g., app1, app2, etc.). \n",
    "\n",
    "However, as noted earlier in the code snippet: `clf = serve.get_deployment_handle(\"image_classifier\", app_name=\"image_classifier_app\")`\n",
    "\n",
    "The app_name must match the name specified in the configuration file. Therefore, you need to update the `applications` section of your YAML file as follows:\n",
    "\n",
    "\n",
    "```\n",
    "applications:\n",
    "  - name: image_classifier_app\n",
    "    import_path: image_classifier:app\n",
    "    route_prefix: /classify\n",
    "    # …other settings…\n",
    "\n",
    "  - name: text_translator_app\n",
    "    import_path: text_translator:app\n",
    "    route_prefix: /translate\n",
    "    # …other settings…\n",
    "\n",
    "  - name: mcp_gateway_app\n",
    "    import_path: mcp_gateway:app\n",
    "    route_prefix: /mcp_gateway\n",
    "    # …other settings…\n",
    "```\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eb53cb77",
   "metadata": {},
   "source": [
    "## 3. Run the service with Ray Serve in terminal\n",
    "\n",
    "We have provied the finalized `config_serve.yaml` file in the folder `mcp-gateway-with-existing-ray-apps/` that include the previous changes, simply run:\n",
    "```bash\n",
    "serve run config_serve.yaml\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6ca61df3",
   "metadata": {},
   "source": [
    "## 4. Test with Python client\n",
    "\n",
    "The service starts on the URL `http://localhost:8000`, because you define the route_prefix: /mcp_gateway on the mcp_gateway_app, therefore, the MCP endpoint becomes `http://localhost:8000/mcp_gateway/mcp`. \n",
    "\n",
    "Note that the endpoint URL adds \"/mcp\" because you have `app.mount(\"/\", mcp.streamable_http_app())` in the `mcp_gateway.py`.\n",
    "   \n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "20e7c0e5",
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio\n",
    "from mcp.client.streamable_http import streamablehttp_client\n",
    "from mcp import ClientSession\n",
    "\n",
    "SAMPLE_IMAGE_URL = \"https://doggos-dataset.s3.us-west-2.amazonaws.com/samara.png\"\n",
    "SAMPLE_TEXT      = \"How are you?\"\n",
    "url = \"http://localhost:8000/mcp_gateway/mcp\"\n",
    "\n",
    "async def main():\n",
    "    async with streamablehttp_client(url=url) as (read, write, _):\n",
    "        async with ClientSession(read, write) as session:\n",
    "            await session.initialize()\n",
    "\n",
    "            # List available tools\n",
    "            tools = await session.list_tools()\n",
    "            print(\"Tools:\")\n",
    "            for t in tools.tools:\n",
    "                print(f\" • {t.name}: {t.description}\")\n",
    "            print()\n",
    "\n",
    "            # Test calls\n",
    "            tests = {\n",
    "                \"classify\": {\"image_url\": SAMPLE_IMAGE_URL},\n",
    "                \"translate\": {\"text\": SAMPLE_TEXT},\n",
    "            }\n",
    "\n",
    "            for t in tools.tools:\n",
    "                if t.name in tests:\n",
    "                    print(f\"--- {t.name} ---\")\n",
    "                    res = await session.call_tool(name=t.name, arguments=tests[t.name])\n",
    "                    for chunk in res.content:\n",
    "                        print(chunk.text)\n",
    "                    print()\n",
    "\n",
    "\n",
    "# ──────── How to run in Jupyter Notebook ────────────────────────────\n",
    "# await main()\n",
    "# ────────────────────────────────────────────────────────────────────\n",
    "\n",
    "\n",
    "# ──────── How to run as a standalone Python script ──────────────────\n",
    "# import asyncio\n",
    "#\n",
    "# if __name__ == \"__main__\":\n",
    "#     # Create and run the event loop\n",
    "#     asyncio.run(main())\n",
    "# ────────────────────────────────────────────────────────────────────\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "479cc657",
   "metadata": {},
   "source": [
    "### To terminate Ray serve:\n",
    "\n",
    "```bash\n",
    "serve shutdown --yes\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "11c0e39d",
   "metadata": {},
   "source": [
    "## 5.  Production deployment with Anyscale service\n",
    "\n",
    "First, copy the `config_serve.yaml` file to a new file named `config_anyscale.yaml`. Then, remove the `proxy_location`, `http_options`, `grpc_options`, and `logging_config` fields. Anyscale services don't support these fields. You can also add the `name` field with the value `mcp-gateway-app-service` to indicate the service name.\n",
    "\n",
    "For production deployment, use Anyscale services to deploy the Ray Serve app to a dedicated cluster. Anyscale ensures scalability, fault tolerance, and load balancing, keeping the service resilient against node failures, high traffic, and rolling updates.\n",
    "\n",
    "Use the following command to deploy the service:\n",
    "\n",
    "```bash\n",
    "anyscale service deploy -f \"config_anyscale.yaml\"\n",
    "```\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "92d52b98",
   "metadata": {},
   "source": [
    "## 6. Query the production service\n",
    "\n",
    "When you deploy, you expose the service to a publicly accessible IP address, which you can send requests to.\n",
    "\n",
    "In the preceding cell’s output, copy your API_KEY and BASE_URL. As an example, the values look like the following:\n",
    "\n",
    "* BASE_URL = \"https://mcp-gateway-app-service-jgz99.cld-kvedzwag2qa8i5bj.s.anyscaleuserdata.com\"\n",
    "* TOKEN = \"CCq8xuiXup_tWcyo-CjfcdyMhiTAnCzQkuXChnmnzoc\"\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "Fill in the following placeholder values for the BASE_URL and API_KEY in the following Python requests object:\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fe73cdc1",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "from mcp.client.streamable_http import streamablehttp_client\n",
    "from mcp import ClientSession\n",
    "\n",
    "BASE_URL = \"https://mcp-gateway-app-service-jgz99.cld-kvedzwag2qa8i5bj.s.anyscaleuserdata.com\" # Replace with your own URL\n",
    "TOKEN = \"CCq8xuiXup_tWcyo-CjfcdyMhiTAnCzQkuXChnmnzoc\" # Replace with your own token\n",
    "url = f\"{BASE_URL}/mcp_gateway/mcp\"\n",
    "\n",
    "# # Common headers for auth\n",
    "headers = {\n",
    "    \"Authorization\": f\"Bearer {TOKEN}\"\n",
    "}\n",
    "\n",
    "async def main():\n",
    "    async with streamablehttp_client(url=url, headers=headers) as (read, write, _):\n",
    "        async with ClientSession(read, write) as session:\n",
    "            await session.initialize()\n",
    "\n",
    "            # List available tools\n",
    "            tools = await session.list_tools()\n",
    "            print(\"Tools:\")\n",
    "            for t in tools.tools:\n",
    "                print(f\" • {t.name}: {t.description}\")\n",
    "            print()\n",
    "\n",
    "            # Test calls\n",
    "            tests = {\n",
    "                \"classify\": {\"image_url\": SAMPLE_IMAGE_URL},\n",
    "                \"translate\": {\"text\": SAMPLE_TEXT},\n",
    "            }\n",
    "\n",
    "            for t in tools.tools:\n",
    "                if t.name in tests:\n",
    "                    print(f\"--- {t.name} ---\")\n",
    "                    res = await session.call_tool(name=t.name, arguments=tests[t.name])\n",
    "                    for chunk in res.content:\n",
    "                        print(chunk.text)\n",
    "                    print()\n",
    "\n",
    "# ──────── How to run in Jupyter Notebook ────────────────────────────\n",
    "# await main()\n",
    "# ────────────────────────────────────────────────────────────────────\n",
    "\n",
    "\n",
    "# ──────── How to run as a standalone Python script ──────────────────\n",
    "# import asyncio\n",
    "#\n",
    "# if __name__ == \"__main__\":\n",
    "#     # Create and run the event loop\n",
    "#     asyncio.run(main())\n",
    "# ────────────────────────────────────────────────────────────────────\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "85331b7a",
   "metadata": {},
   "source": [
    "## 7. Test with MCP Inspector"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1b0080d1",
   "metadata": {},
   "source": [
    "Follow the instructions in Notebook #1 to set up MCP Inspector on your local machine. \n",
    "\n",
    "Then go to the `http://127.0.0.1:6274/` to find the Inspector UI and enter the following: \n",
    "\n",
    "* Transport Type: Streamable HTTP\n",
    "* URL: https://mcp-gateway-app-service-jgz99.cld-kvedzwag2qa8i5bj.s.anyscaleuserdata.com/mcp_gateway/mcp\n",
    "* Bearer token = \"CCq8xuiXup_tWcyo-CjfcdyMhiTAnCzQkuXChnmnzoc\"\n",
    "\n",
    "Note: Use your own service URL and bearer token.\n",
    "\n",
    "Then you can test the following remote MCP servers that have connected with your Ray serve apps. \n",
    "\n",
    "\n",
    "### Image classifier tool calling result:\n",
    "\n",
    "<img\n",
    "  src=\"https://agent-and-mcp.s3.us-east-2.amazonaws.com/mcp/diagrams/mcp_gateway_demo1.png\"\n",
    "  alt=\"MCP Gateway Demo1\"\n",
    "  style=\"width:80%;\"\n",
    "/>\n",
    "\n",
    "\n",
    "### Text translator tool calling result:\n",
    "<img\n",
    "  src=\"https://agent-and-mcp.s3.us-east-2.amazonaws.com/mcp/diagrams/mcp_gateway_demo2.png\"\n",
    "  alt=\"MCP Gateway Demo2\"\n",
    "  style=\"width:80%;\"\n",
    "/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "59b2207d",
   "metadata": {},
   "source": [
    "\n",
    "## 8. Terminate the Anyscale Service\n",
    "After testing the service, you can shutdown the serive with this command:\n",
    "```bash\n",
    "anyscale service terminate --name=mcp-gateway-app-service\n",
    "```"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "ray-doc-mcp-ray-serve",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
